package wc;
 
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Collection;
import java.util.Set;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
 
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
 
public class FriendRecom {
 
	static class ValuePair implements WritableComparable<ValuePair> {
 
		private String friend;
		private String union;
 
		public String getFriend() {
			return friend;
		}
 
		public void setFriend(String friend) {
			this.friend = friend;
		}
 
		public String getUnion() {
			return union;
		}
 
		public void setUnion(String union) {
			this.union = union;
		}
 
		public ValuePair() {
			// TODO Auto-generated constructor stub
		}
 
		public ValuePair(String n, String u) {
			this.friend = n;
			this.union = u;
		}
 
		@Override
		public void readFields(DataInput in) throws IOException {
			this.friend = in.readUTF();
			this.union = in.readUTF();
		}
 
		@Override
		public void write(DataOutput out) throws IOException {
			out.writeUTF(this.friend);
			out.writeUTF(this.union);
		}
 
		@Override
		public int compareTo(ValuePair o) {
			if (this.friend.equals(o.getFriend())) {
				return this.union.compareTo(o.getUnion());
			} else {
				return this.friend.compareTo(o.getFriend());
			}
		}
 
	}

	public static class FirstMapper extends Mapper<Object, Text, Text, Text> {

        @Override
        public void map(Object keys, Text values, Context context) throws IOException, InterruptedException {
            String val = values.toString();
            String[] split = val.split("\t");
            String key = split[0];
            String value = split[1];
            context.write(new Text(key), new Text(value));
        }
    }

    public static class FirstReducer extends Reducer<Text, Text, Text, Text> {

        public void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException {
            StringBuffer stringBuffer = new StringBuffer();
            for (Text text : values){
                stringBuffer.append(text).append(",");
            }
            context.write(key,new Text(stringBuffer.toString().substring(0,stringBuffer.toString().length()-1)));
        }
    }



 
	public static class Map extends Mapper<LongWritable, Text, Text, ValuePair> {
 
		private Text outkey = new Text();
		private ValuePair outvalue = new ValuePair();
 
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String input = value.toString();
			String[] sz = input.split("\t");
			this.outkey.set(sz[0]);
			String[] sz1 = sz[1].split(",");
			for (int i = 0; i < sz1.length; i++) {
				this.outvalue.setFriend(sz1[i]);
				this.outvalue.setUnion("yes"); 
				context.write(this.outkey, this.outvalue);
			}
 
			for (int i = 0; i < sz1.length; i++) {
				for (int j = i + 1; j < sz1.length; j++) {
					this.outkey.set(sz1[i]);
					this.outvalue.setFriend(sz1[j]);
					this.outvalue.setUnion(sz[0]); 
					context.write(this.outkey, this.outvalue);
 
					this.outkey.set(sz1[j]);
					this.outvalue.setFriend(sz1[i]);
					context.write(this.outkey, this.outvalue);
				}
			}
		}
	}
 
	public static class Reduce extends Reducer<Text, ValuePair, Text, Text> {
 
		private Text outKey = new Text();
		private Text outValue = new Text();
 
		@Override
		protected void reduce(Text key, Iterable<ValuePair> values,
				Context context) throws IOException, InterruptedException {
			this.outKey = key;
			Multimap<String, String> map = HashMultimap.create();
			for (ValuePair v : values) {
				map.put(v.getFriend(), v.getUnion()); 
			}
 
			StringBuilder outString = new StringBuilder();
			Set<String> keys = map.keySet();
			for(String s : keys){
				StringBuilder outStr = new StringBuilder();
				boolean isok = true;
				Collection<String> v = map.get(s);
				outStr.append(s + "(");
				String tmp = new String();
				for(String u : v){
					if(u.equals("yes")){
						isok = false;
						break;
					}
					tmp += u + ", ";
				}
				if(tmp.length()>2){
					tmp = tmp.substring(0, tmp.length()-2);
				}
				if(isok){
					outStr.append(v.size() + ": " + "[" + tmp.trim() + "]),");
					outString.append(outStr);
				}
			}
			
			outValue.set(outString.toString());
			context.write(outKey, outValue);
			
		}
 
	}

 
	public static void main(String[] args) throws Exception {
		Configuration conf0 = new Configuration();
        Job job0 = Job.getInstance(conf0, "friend 1.0");
        job0.setJarByClass(FriendRecom.class);
        job0.setMapperClass(FirstMapper.class);
        job0.setReducerClass(FirstReducer.class);
        job0.setOutputKeyClass(Text.class);
        job0.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job0, new Path(args[0]));
        FileOutputFormat.setOutputPath(job0, new Path("tmp1"));
        job0.waitForCompletion(true);



		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf0, "friend 2.0");
		job.setJarByClass(FriendRecom.class);
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(ValuePair.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		FileSystem fs = FileSystem.get(conf);
		Path inPath = new Path("tmp1");
		if (fs.exists(inPath)) {
			FileInputFormat.addInputPath(job, inPath);
		}
 
		Path outPath = new Path(args[1]);
		fs.delete(outPath, true);
		FileOutputFormat.setOutputPath(job, outPath);
 
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
 
}